MediaLiveでの配信終了時に自動でアーカイブファイルを変換する仕組みを作ってみた
こんにちは、大前です。
MediaLive での配信終了時に自動で MediaConvert の変換ジョブを走らせる仕組みを作ってみました。
構成図
構成は以下の通りです。
- MediaLive から S3 にアーカイブを出力するように設定
- CloudWatch Events で MediaLive チャネルのステータスを監視
- MediaLive チャネルが「停止」になったら指定の Lambda 関数を呼び出す
- Lambda 関数から MediaConvert の変換ジョブをリクエスト
- MediaConvert がアーカイブファイルに対して変換ジョブを実行
やってみた
実際にやってみます。
1. MediaLive チャネルを作成
まずは、MediaLive の設定をしていきます。チャネル作成時の詳細な手順を確認したい場合はこちらを参照ください。
チャネル名を入力し、ロールを選択
入力をアタッチ。今回は OBS からテストするので RTMP プッシュの入力を作っておきました。
出力にアーカイブグループを追加。実際に運用する時は、この他に配信用の出力を作成する形となります。
アーカイブの送信先となる S3 を指定。
出力設定は以下。
ここまで設定したら、「作成」を押下します。
2. MediaConvert でジョブテンプレートを作成
MediaConvert でジョブテンプレートを作成します。
ジョブテンプレートの作成から、以下のようなテンプレートを作成しました。
入力は Lambda 側で設定する為、テンプレートでは何も設定しないのがポイントです。
出力は、よしなに設定してください。
3. Lambda 関数を作成
続いて、MediaConvert のジョブを呼び出す Lambda 関数を作成します。
作成にあたっては、こちらを参考にしました。
ランタイムは Python 3.7、ロールはとりあえずデフォルトで作成します。
実行ロールに、以下のポリシーをアタッチします。
MediaConvertRole の中身は以下です。
{ "Version": "2012-10-17", "Statement": [ { "Sid": "VisualEditor0", "Effect": "Allow", "Action": [ "iam:PassRole", "mediaconvert:CreateJob" ], "Resource": "*" } ] }
ソースは以下になります。
import json import urllib.parse import boto3 import os MEDIACONVERT_ENDPOINT = os.environ['MEDIACONVERT_ENDPOINT'] # MediaConvertのエンドポイント TARGET_BUCKET = os.environ['TARGET_BUCKET'] # アーカイブファイルが格納されているバケット TARGET_BUCKET_PREFIX = os.environ['TARGET_BUCKET_PREFIX'] # アーカイブファイルのキー(フォルダ名) JOB_TEMPLATE = os.environ['JOB_TEMPLATE'] # MediaConvertのジョブテンプレート QUEUE = os.environ['QUEUE'] # MediaConvertのキューARN ROLE = os.environ['ROLE'] # MediaConvertの実行ロール def lambda_handler(event, context): # クライアント初期化 s3 = boto3.client('s3') mediaconvert = boto3.client('mediaconvert', region_name='ap-northeast-1', endpoint_url=MEDIACONVERT_ENDPOINT) # アーカイブファイルのリスト取得 targetlist = get_targetlist(s3) if len(targetlist) == 0: # ファイルが見つからなかったら終了 return # MediaConvertのリクエスト生成 input_objects = [] for target in targetlist: with open("input_template.json", "r") as jsonfile: input_object = json.load(jsonfile) input_object["FileInput"] = "s3://" + TARGET_BUCKET + "/" + target input_objects.append(input_object) print('input_objects : ' + str(input_objects)) with open("job.json", "r") as jsonfile: job_object = json.load(jsonfile) job_object["Inputs"] = input_objects print('job_object : ' + str(job_object)) # リクエスト実行 response = mediaconvert.create_job( JobTemplate = JOB_TEMPLATE, Queue = QUEUE, Role = ROLE, Settings=job_object ) return { 'statusCode': 200 } def get_targetlist(s3): # オブジェクト取得 res = s3.list_objects_v2( Bucket = TARGET_BUCKET, Prefix = TARGET_BUCKET_PREFIX ) result_list = [] for obj in res['Contents']: # list_objectsでフォルダ名のみの文字列も返却されるため弾く if obj['Key'].split('/')[1] != '': result_list.append(obj['Key']) print('result_list : ' + str(result_list)) return result_list
処理の概要としては、以下になります。
- アーカイブファイルが格納されている S3 バケットから変換対象のファイル一覧を取得
- MediaConvert のリクエストを作成
- MediaConvert にジョブのリクエストを実行
細かいエラーハンドリング等はしていないので、引用する場合には適宜改修してお使いください。
ソース内で指定しているファイルは、以下になります。
MediaConvert で作成したジョブテンプレートの JSON を取得出来るので、それを元にして動的に入力を設定する為のファイルを作成しました。
input_template.json
{ "AudioSelectors": { "Audio Selector 1": { "Offset": 0, "DefaultSelection": "DEFAULT", "ProgramSelection": 1 } }, "VideoSelector": { "ColorSpace": "FOLLOW", "Rotate": "DEGREE_0", "AlphaBehavior": "DISCARD" }, "FilterEnable": "AUTO", "PsiControl": "USE_PSI", "FilterStrength": 0, "DeblockFilter": "DISABLED", "DenoiseFilter": "DISABLED", "TimecodeSource": "EMBEDDED", "FileInput": "" }
job.json
{ "Inputs":[] }
こんな感じで 3つのファイルが存在する形となります。
4. CloudWatch Events を設定
最後に、CloudWatch Events を以下の様に設定しました。
resources には、作成した MediaLive のチャネル ARN を指定してください。
5. 動かしてみる
MediaLive のチャネルを開始し、OBS でチャネルに対して配信を開始します。
チャネルを停止させます
停止完了
MediaConvert のジョブを見に行くと、生成されたアーカイブファイル全てを入力としてジョブが実行されている事を確認できます。
また、S3 にも変換後のファイルが生成されていました。
おわりに
MediaLive のアーカイブを配信停止時に自動変換する仕組みを作ってみました。
今までは配信後に手動で MediaConvert のジョブを作成していたのですが、その手間が省けたので満足しています。
以上、AWS 事業本部の大前でした。